package org.example;



import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.TypeReference;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.example.datasource.impl.DataLinkMedataDataSourceImpl;
import org.example.datasource.model.FieldMetadata;
import org.example.datasource.model.TableMetadata;
import org.example.dolph.DolphinschedulerClient;
import org.example.dolph.model.ProjectInfo;
import org.example.dolph.model.TaskDefine;
import org.example.dolph.model.TaskDetail;
import org.example.model.ColumnLineage;
import org.example.model.DsDatasourceInfo;
import org.example.model.TableLineage;
import org.example.seatunnel.SeatunnelSinkWrapper;
import org.example.seatunnel.SeatunnelSourceWrapper;
import org.example.seatunnel.model.SinkResModel;
import org.example.seatunnel.model.SourceResModel;

import java.util.*;
import java.util.stream.Collectors;

public class SeatunnelParserv2 {


    public List<TableLineage> parse(String sql) {
        return null;
    }

    public static void main(String[] args) {


        String token = "d7516af061cd2af12e9f866f2a4199c9";	// dolphin scheduler token
        String dolphinAddress = "http://172.30.33.173:32777/dolphinscheduler";  // dolphin scheduler address
        // 元数据地址
        MetadataClient metadataClient = new MetadataClient("http://172.30.33.81:9000");
        String pyAddress = "http://172.30.33.81:8000/api/lineage";

        DataLinkMedataDataSourceImpl.address = "http://172.30.33.81:9000";

       // seatunnelParser.parse();



        DolphinschedulerClient dolphinschedulerClient = new DolphinschedulerClient(token, dolphinAddress);

        for (ProjectInfo projectInfo : dolphinschedulerClient.getProjects().getTotalList()) {
            int count = 0;
            int pushCount = 0;
            for (TaskDefine taskDefine : dolphinschedulerClient.getTaskDefines(projectInfo.getCode()).getTotalList()) {
                boolean isDo = true;

                if(taskDefine.getTaskType().equals("SQL")){
                    TaskDetail taskDefine1 =  dolphinschedulerClient.getTaskDetail(projectInfo.getCode(),taskDefine.getTaskCode());
                    JSONObject jsonObject = JSON.parseObject(taskDefine1.getTaskParams());
                    String sql = jsonObject.getString("sql");
                    if (sql.startsWith("insert")){
                        DsDatasourceInfo dsDatasourceInfo =  dolphinschedulerClient.getDsDataSource(jsonObject.getInteger("datasource"));
                        sql = sql.replaceAll("\\[","");
                        sql = sql.replaceAll("]","");
                        sql = sql.replaceAll("(?<!\\.)`", " ").replaceAll("\\.`", ".");
                        System.out.println(sql);
                        metadataClient.pushSql(dsDatasourceInfo.getHost(),dsDatasourceInfo.getPort(),sql);
                    }
                }else {
                    continue;
                }
                if(taskDefine.getTaskType().equals("SEATUNNEL")){
                    TaskDetail taskDefine1 =  dolphinschedulerClient.getTaskDetail(projectInfo.getCode(),taskDefine.getTaskCode());

                    Config config =  ConfigFactory.parseString(taskDefine1.getSeaTaskParams().getRawScript());

                    Config configValues =  config.getConfig("source");

                    try{
                        SourceResModel sourceResModel =  SeatunnelSourceWrapper.parse(configValues);
                        if (sourceResModel != null){
                            Config sink =  config.getConfig("sink");
                            String url = sink.getConfig("jdbc").getString("url");
                            count ++;

                            SinkResModel sinkRes = SeatunnelSinkWrapper.parse(sourceResModel,sink);



                            Map<String,List<String>> metadata = new HashMap<>();
                            for (TableMetadata tableMetadata : sinkRes.getSourceMetadataList()) {
                                List<String> fields = tableMetadata.getFields().stream().map(FieldMetadata::getFileName).collect(Collectors.toList());
                                metadata.put("source."+tableMetadata.getTableName(),fields);
//                                tableMetadata.getDatabaseName()
                            }
                            JSONObject jsonObject = new JSONObject();
                            jsonObject.put("metadata",metadata);
                            jsonObject.put("sql",sinkRes.getSql());
                            String res = HttpUtil.post(pyAddress,jsonObject.toJSONString());
                            Map<String,TableLineage> tableLineageMap = new HashMap<>();

                            List<List<JSONObject>> resList = JSON.parseObject(res,new TypeReference<List<List<JSONObject>>>() {});
                            for (List<JSONObject> strings : resList) {
//                                System.out.println(strings);
                                JSONObject sourceTable = strings.get(0).getJSONArray("_parent").getJSONObject(0);
                                String sourceTableSchema = sourceTable.getJSONObject("schema").getString("raw_name");
                                String sourceTableName = sourceTable.getString("raw_name");
                                String sourceField = strings.get(0).getString("raw_name");


                                JSONObject targetTable = strings.get(1).getJSONArray("_parent").getJSONObject(0);
                                String  targetTableSchema = targetTable.getJSONObject("schema").getString("raw_name");
                                String  targetTableName = targetTable.getString("raw_name");
                                String  targetField = strings.get(1).getString("raw_name");
                                Optional<TableMetadata> tb1 =  sinkRes.getSinkMetadataList().stream()
                                        .filter(i->i.getTableName().equalsIgnoreCase(targetTableName)).findFirst();
                                if (tb1.isEmpty()){
                                    isDo = false;
                                }
                                TableLineage tableLineage =  tableLineageMap.computeIfAbsent(targetTableSchema + "." + targetTableName,s->{
                                    TableLineage tableLineage1 = new TableLineage();
//                                    tableLineage1.setDatabaseName(targetTableSchema);
                                    tableLineage1.setTableName(targetTableName);
                                    Optional<TableMetadata> tb =  sinkRes.getSinkMetadataList().stream()
                                            .filter(i->i.getTableName().equalsIgnoreCase(targetTableName)).findFirst();
                                    if (tb.isPresent()){
                                        tableLineage1.setIp(tb.get().getIp());
                                        tableLineage1.setPort(tb.get().getPort());
                                        tableLineage1.setDatabaseName(tb.get().getDatabaseName());
                                    }else {
                                        for (TableMetadata tableMetadata : sinkRes.getSinkMetadataList()) {
                                            System.out.println(tableMetadata.getTableName());
                                        }
                                        System.out.println("sink 未找到："+ targetTableName);
                                        System.out.println(sink);


                                    }
                                    return tableLineage1;
                                });


                                List<ColumnLineage> columnLineages =
                                        tableLineage.getColumnLineages().computeIfAbsent(targetField,s -> new ArrayList<>());

                                ColumnLineage columnLineage = new ColumnLineage();
                                columnLineage.setSourceColumn(sourceField);
                                TableLineage sourceTableInfo = new TableLineage();
//                                sourceTableInfo.setDatabaseName(targetTableSchema);
                                sourceTableInfo.setTableName(sourceTableName);
                                Optional<TableMetadata> tb =  sinkRes.getSourceMetadataList().stream()
                                        .filter(i->i.getTableName().equalsIgnoreCase(sourceTableName)).findFirst();
                                if (tb.isPresent()){
                                    sourceTableInfo.setIp(tb.get().getIp());
                                    sourceTableInfo.setPort(tb.get().getPort());
                                    sourceTableInfo.setDatabaseName(tb.get().getDatabaseName());

                                }else {
                                    for (TableMetadata tableMetadata : sinkRes.getSourceMetadataList()) {
                                        System.out.println(tableMetadata.getTableName());
                                    }
                                    System.out.println("souce 未找到："+ sourceTableName);
                                    System.out.println(configValues);
                                    isDo = false;                                }
                                columnLineage.setSourceTable(sourceTableInfo);
                                columnLineages.add(columnLineage);
                            }
                            if (isDo){
//                            for (TableLineage value : tableLineageMap.values()) {
//                                value.get
//                            }
                                if (tableLineageMap.values().stream().findFirst().get().getDatabaseName().equalsIgnoreCase("etl_ods")){
                                    System.out.println("已经推送到元数据系统：" + taskDefine.getTaskName());
                                    System.out.println(JSON.toJSONString(tableLineageMap.values().stream().findFirst().get().getDatabaseName()));
                                    pushCount++;
                                    System.out.println(JSON.toJSONString(tableLineageMap.values()));
                                    metadataClient.sub(new ArrayList<>(tableLineageMap.values()));

                                }

                            }else {
                                System.out.println("失败");
                            }



                        }
                    }catch (Exception e){
                        System.out.println(taskDefine.getTaskName()+" : 解析失败");
//                        System.out.println(config);
                        e.printStackTrace();
                    }

                }
            }
            System.out.println("项目名：" + projectInfo.getName() + "数量："+count + " 推送数量:" + pushCount);
        }



    }
}



//#Defining the runtime environment
//env {
//  # You can set flink configuration here
//    execution.parallelism = 4
//    job.mode = "BATCH"
//}
//source{
//    Jdbc {
//        url = "jdbc:sqlserver://172.30.36.154:1433;databaseName=SurveyAndDesign;encrypt=true;trustServerCertificate=true"
//        driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
//        user = "reader1"
//        password = "abc123!@#"
//        query ="select id,ryid,byyx,sxzy,rxsj,bysj,bz,status,create_time,zzcl from t_gd_ry_xxjl;"
//    }
//}
//
//sink {
//    jdbc {
//        url = "jdbc:mysql://172.30.36.144:9031/etl_ods?rewriteBatchedStatements=true&useUnicode=true&characterEncoding=utf-8"
//        driver = "com.mysql.cj.jdbc.Driver"
//        user = "root"
//        password = ""
//        query = "insert into t_gd_ry_xxjl(id,ryid,byyx,sxzy,rxsj,bysj,bz,status,create_time,zzcl) VALUES ( ?,?,?,?,?,?,?,?,?,? )"
//    }
//}